package com.gulimall.search.thread;

import ch.qos.logback.classic.util.LogbackMDCAdapter;
import com.google.common.io.Files;
import org.apache.commons.codec.Charsets;
import org.apache.commons.io.FileUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

/**
 * 单线程处理大文件
 */
public class IoTest {
    private static final String ENCODING = "utf-8";
    //处理文件段大小，默认20000
    private static final int pageSize = 20000;
    //读取当前CPU个数，决定线程池的大小
    private final static int POOLSIZE = Runtime.getRuntime().availableProcessors();
    //线程池
    private final static ExecutorService executor = Executors.newFixedThreadPool(POOLSIZE);
    //任务队列
    private final static ArrayBlockingQueue<List<String>> taskQueue = new ArrayBlockingQueue<>(POOLSIZE);

    private final static AtomicBoolean isFinsh = new AtomicBoolean(false);

    public static void main(String[] args) throws Exception {

        File file = new File("D:\\test_new.csv");
        if (!file.exists()) {
            create();
        }
       /* long start = System.currentTimeMillis();
        Files.readLines(new File("D:\\test.txt"), Charsets.UTF_8);
        System.out.println("总耗时：" + (System.currentTimeMillis() - start) + "毫秒");*/

        //单线程处理
        /* handle(file);*/
        // 多线程处理
        long start = System.currentTimeMillis();
        handleSync(file);
        System.out.println("总耗时：" + (System.currentTimeMillis() - start) + "毫秒");
    }

    /**
     * 执行(单线程)
     *
     * @param file 文件
     * @throws IOException 错误
     */
    public static void handle(File file) throws IOException {
        int index = 1;
        long start = System.currentTimeMillis();
        List<String> lines = copyToLines(file, "utf-8", index, 20000);
        while (!CollectionUtils.isEmpty(lines)) {
            /* handleInternal(lines, index);*/
            System.out.println("正在处理" + index + "-" + (index + 20000 - 1) + "处数据...");
            index = index + 20000;
            lines.clear();
            lines = copyToLines(file, "utf-8", index, 20000);
        }
        System.out.println("总耗时：" + (System.currentTimeMillis() - start) + "毫秒");
    }

    /**
     * 读取文件
     *
     * @param file      文件
     * @param encoding  编码
     * @param startLine 开始位置
     * @param size      大小
     * @return 字符串
     * @throws IOException 错误
     */
    public static List<String> copyToLines(File file, String encoding, int startLine,
                                           int size) throws IOException {
        List<String> lines = new ArrayList<String>();
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file), encoding))) {
            String line;
            int lineNum = 0;
            if (startLine < 1) startLine = 1;
            while ((line = reader.readLine()) != null) {
                lineNum++;
                if (lineNum < startLine) continue;
                if (size > 0 && (lineNum - startLine) >= size) break;
                lines.add(line);
            }
        }
        return lines;
    }

    private static void handleInternal(List<String> lines) {
        File file = new File("D:\\test_new.csv");
        StringBuilder sb = new StringBuilder();
        for (String line : lines) {
            sb.append(line).append("\n");
        }
        //直接写成新文件
        write(file, sb.toString());
    }

    protected static void write(File file, String str) {
        try {
            FileUtils.writeStringToFile(file, str, "utf-8", true);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建文件
     */
    public static void create() throws Exception {
        final String json = "[{\"id\":\"380001671739949056\",\"type\":\"3\",\"work_no\":\"chn001\",\"display_name\":\"测试chn\",\"display_py\":\"ceschn\",\"country\":\"1001\",\"id_card_no\":null,\"birthday\":null,\"sex\":null,\"secret_level\":\"1\",\"nationality\":\"1\",\"mobile\":null,\"sw_email\":null,\"ip_phone\":null,\"leave_date\":null,\"marital_status\":null,\"certificate_cn\":null,\"politics_status\":null,\"native_place\":null,\"work_time\":null,\"image\":null,\"status\":1,\"update_time\":1631601059000,\"create_time\":1631601076000,\"userJobInfo\":[{\"company_id\":\"129343111846100992\",\"company_job_type\":\"0\",\"jh_email\":null,\"email\":null,\"office_address\":null,\"job_level\":null,\"depJobInfo\":[{\"org_id\":\"129343111846100992\",\"dept_job_type\":\"1\",\"job\":null,\"job_status\":null,\"position\":null,\"show_order\":null,\"valid_time\":null,\"job_attribute\":null}]}],\"syncStatus\":2000,\"syncSequence\":380001673615736833},{\"id\":\"380388589673725952\",\"type\":\"1\",\"work_no\":\"727411\",\"display_name\":\"冬测试导入用户-11\",\"display_py\":\"dongcsdryh-11\",\"country\":null,\"id_card_no\":null,\"birthday\":null,\"sex\":null,\"secret_level\":\"1\",\"nationality\":null,\"mobile\":null,\"sw_email\":null,\"ip_phone\":null,\"leave_date\":null,\"marital_status\":null,\"certificate_cn\":null,\"politics_status\":null,\"native_place\":null,\"work_time\":null,\"image\":null,\"status\":1,\"update_time\":1631693308000,\"create_time\":1631693324000,\"userJobInfo\":[{\"company_id\":\"114836879844122624\",\"company_job_type\":\"0\",\"jh_email\":null,\"email\":null,\"office_address\":null,\"job_level\":null,\"depJobInfo\":[{\"org_id\":\"124944489058598912\",\"dept_job_type\":\"1\",\"job\":null,\"job_status\":null,\"position\":null,\"show_order\":null,\"valid_time\":null,\"job_attribute\":null}]}],\"syncStatus\":2000,\"syncSequence\":380388591876669441}]";
        final BufferedWriter writer = new BufferedWriter(new FileWriter("D:\\test.txt"));
        long start = System.currentTimeMillis();
        IntStream.range(0, 800000).forEach(value -> {
            try {
                //新起一行 写数据
                writer.newLine();
                writer.write(json);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        long end = System.currentTimeMillis();
        writer.close();
        System.out.println((end - start) / 1000 + "ms");
    }

    /**
     * 把文件拆分成size大小分别处理，多线程
     *
     * @param file 文件
     */
    public static void handleSync(File file) {
        //将大文件拆分后放到任务队列
        executor.execute(() -> {
            try {
                putTasks(file);
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
            }
        });
        //处理业务
        handleSyncInternal();
        executor.shutdown();

        System.out.println("==任务执行完毕==");
    }

    /**
     * 功能: 多线程处理具体任务
     */
    private static void handleSyncInternal() {
        try {
            AtomicInteger startIndex = new AtomicInteger(1);
            while (!isFinsh.get()) {
                List<String> lines = taskQueue.poll(3,TimeUnit.MICROSECONDS);
                if (CollectionUtils.isEmpty(lines))
                    continue;
                executor.execute(() -> {
                    int index = startIndex.getAndAdd(pageSize);
                    System.out.println("正在处理" + index + "-" + (index + pageSize - 1) + "处数据...");
                    //handleInternal(lines);
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * <p>功能: 往任务队列里放入任务</p>
     *
     * @param file 文件
     * @throws IOException          io错误
     * @throws InterruptedException 输入错误
     */
    private static void putTasks(File file) throws IOException, InterruptedException {
        List<String> lines = new ArrayList<String>();
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file), ENCODING))) {
            String line = null;
            while ((line = reader.readLine()) != null) {
                //忽略空行
                if (!StringUtils.hasText(line)) continue;
                lines.add(line);
                //如果list大于pageSize放入任务队列
                if (lines.size() >= pageSize) {
                    taskQueue.put(lines);
                    lines = new ArrayList<String>();
                }
            }
            //放入最后剩余部分
            taskQueue.put(lines);
        } finally {
            System.out.println("==生产线程结束了==");
            isFinsh.set(true);
        }
    }

}
